package com.example.push.service;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

@Service
@Slf4j
@AllArgsConstructor
public class KafkaProducerService {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private static final String TOPIC = "push-topic";

    public void sendMessage(String message) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC, message);

        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(@Nullable SendResult<String, String> result) {
                if (result != null) {
                    log.info("Message sent successfully to topic {} with offset {}",
                        result.getRecordMetadata().topic(),
                        result.getRecordMetadata().offset());
                } else {
                    log.warn("Message sent but result is null");
                }
            }

            @Override
            public void onFailure(@NonNull Throwable ex) {
                log.error("Failed to send message to topic {}: {}", TOPIC, ex.getMessage());
            }
        });
    }
}